#!/usr/bin/python
# -*- coding: utf-8 -*-  
"""
@Project : hello 
@file : flink_stream_job.py
@Author : shenj
@time : 2025/4/2 16:56
@func : 运行中的flink-crmbo-job的监控和重启
"""

import pymysql
import requests

from com.cn.for_cdc.common.cdc_conmons import mysql_pro_connections


def get_running_jobs(bu, url):
    response = requests.get(url)
    # 检查请求是否成功
    if response.status_code == 200:
        try:
            data = response.json()
            running_jobs = filter(lambda x: x['state'] == 'RUNNING', data['jobs'])
            name_jid_dict = {job['name']: job['jid'] for job in running_jobs}
            print(f"{bu}正在运行的stream-crmbo-job数量:{len(name_jid_dict)}个")
            # print(f"{bu}运行的任务:{name_jid_dict}")
            return name_jid_dict
        except ValueError:
            print("Response content is not valid JSON")
    else:
        print(f"Failed to retrieve data: {response.status_code}")


def get_connection():
    db_configs = mysql_pro_connections[f'{bu}_pro_job']
    conn = pymysql.connect(**db_configs)
    return conn


def get_flink_monitoring_job(bu):
    try:
        # 定义查询语句
        query_sql = f"select job_name from asw_etl_{bu}_job.FLINK_MONITORING_JOB ;"
        conn = get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute(query_sql)
                datas = cursor.fetchall()
                result_list = [item[0] for item in datas]
                return result_list
        except Exception as e:
            print("数据库操作异常：\n", e)
        finally:
            conn.close()
    except Exception as e:
        print(f"An error occurred: {e}")


def find_unmonitored_tasks(all_tasks_dict, monitored_list):
    return list(set(all_tasks_dict.keys()) - set(monitored_list))


def add_monitor_job(bu,url,job_name):
    # 给flink实时任务加监控
    insert_sql = f"""INSERT
                        INTO
                        asw_etl_{bu}_job.FLINK_MONITORING_JOB
                    (PK,
                        ADDRESS,
                        JOB_NAME,
                        `TYPE`,
                        STATUS,
                        CREATED_DATE,
                        UPDATED_DATE)
                    VALUES((SELECT MAX(pk) + 1 FROM (SELECT * FROM asw_etl_{bu}_job.FLINK_MONITORING_JOB) AS temp),
                    '{url}',
                    '{job_name}',
                    '3',
                    '1',
                    CURRENT_TIMESTAMP(),
                    CURRENT_TIMESTAMP());
                    """
    conn = get_connection()  # 创建数据库连接
    try:
        with conn.cursor() as cursor:
            cursor.execute(insert_sql)
            conn.commit()
            print(f"添加监控任务:{job_name}")
            print(f"添加监控任务的SQl:{insert_sql}")
    except Exception as e:
        print("数据库操作异常：\n", e)
    finally:
        conn.close()


def find_unrestart_tasks(all_tasks_dict, restarted_list):
    all_tasks_upper = {k.upper() for k in all_tasks_dict}
    restarted_upper = {task.upper() for task in restarted_list}
    return list(all_tasks_upper - restarted_upper)


def get_xx_schedule_job(bu):
    query_sql = f"select job_name from asw_etl_{bu}_job.XX_SCHEDULE_JOB where INSTANCE_NAME='flinkOutboundJob' -- and job_name like '%CRMBO%' ;"
    try:
        # 定义查询语句
        conn = get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute(query_sql)
                datas = cursor.fetchall()
                result_list = [item[0] for item in datas]
                return result_list
        except Exception as e:
            print("数据库操作异常：\n", e)
        finally:
            conn.close()
    except Exception as e:
        print(f"An error occurred: {e}")


def get_xx_schedule_job_remark(bu):
    try:
        # 定义查询语句
        query_sql = f"SELECT JSON_EXTRACT(PARAM_JSON, '$.jobName') as name,remark FROM asw_etl_{bu}_job.XX_SCHEDULE_JOB  WHERE PARAM_JSON like '%cdc-{bu}-outbound-stream-crmbo%' ;"
        # print(query_sql)
        conn = get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute(query_sql)
                datas = cursor.fetchall()
                name_remark_dict = {row[0].replace('"', ''): row[1] for row in datas if row[0]}
                return name_remark_dict
        except Exception as e:
            print("数据库操作异常：\n", e)
        finally:
            conn.close()
    except Exception as e:
        print(f"An error occurred: {e}")


def excute_update_sql(sql):
    conn = get_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            conn.commit()
            return cursor.rowcount
    except pymysql.Error as e:
        conn.rollback()
        raise RuntimeError(f"Execute failed: {e}")
    finally:
        cursor.close()
        conn.close()


def compare_dict(bu, all_tasks_dict, name_remark_dict):
    unrestart_job_list = []
    err_remark_list = []
    # job_name 不一致校验
    for job_name, jid in all_tasks_dict.items():
        if job_name not in name_remark_dict:
            print(f"{job_name} 未配置重启")
            unrestart_job_list.append(job_name)
        # else:
        #     print(f"{job_name}已配置重启")

    # jid不一致校验
    for job_name, jid in all_tasks_dict.items():
        if job_name in name_remark_dict:
            if jid != name_remark_dict[job_name]:
                # print(f"{job_name} 的remark错误，jid:{jid}，数据库中remark为:{name_remark_dict[job_name]}")
                # update_sql=f"update asw_etl_{bu}_job.XX_SCHEDULE_JOB set remark='{jid}' where JSON_EXTRACT(PARAM_JSON, '$.jobName')='{job_name}' ;"
                update_sql = f"""
                            update asw_etl_{bu}_job.XX_SCHEDULE_JOB set remark='{jid}' 
                            WHERE
                                PARAM_JSON IS NOT NULL
                                AND PARAM_JSON <> ''
                                AND JSON_VALID(PARAM_JSON)
                                AND JSON_EXTRACT(PARAM_JSON, '$.jobName') = '{job_name}';"""
                # print(update_sql)
                # excute_update_sql(update_sql)

    print(f"{bu}未配置重启的任务数量:{len(unrestart_job_list)}个")
    print(f"{bu}配置重启的remark错误的任务数量:{len(err_remark_list)}个")
    return unrestart_job_list, err_remark_list


def deal_restart_task(all_tasks_dict, bu):
    # restarted_list = get_xx_schedule_job(bu)
    # unrestarted_jobs = find_unrestart_tasks(all_tasks_dict, restarted_list)
    # print(f"{bu}未配置重启的任务数量:{len(unrestarted_jobs)}个")
    # print(unrestarted_jobs)

    # 对比重启任务的jb_name,job_id
    name_remark_dict = get_xx_schedule_job_remark(bu)

    compare_dict(bu, all_tasks_dict, name_remark_dict)
    # print(all_tasks_dict)
    # print(name_remark_dict)


def deal_monitor_task(bu, all_tasks_dict, url):
    # 获取已经监控的任务
    monitored_list = get_flink_monitoring_job(bu)
    # # 找出未监控的任务
    unckeck_jobs = find_unmonitored_tasks(all_tasks_dict, monitored_list)
    print(f"未添加监控的任务数量:{len(unckeck_jobs)}个")
    # 将未监控的任务添加到数据库中
    if unckeck_jobs:
        print(f"{bu}未监控的任务:", unckeck_jobs)
        # for job_name in unckeck_jobs:
        #     add_monitor_job(bu,url,job_name)


def genert_sql(bu, all_tasks_list):
    query_sql = f"""
                SELECT
                    *
                FROM
                    asw_etl_{bu}_job.XX_SCHEDULE_JOB
                WHERE
                    PARAM_JSON like '%cdc-{bu}-outbound-stream-crmbo%'
                    and JSON_EXTRACT(PARAM_JSON, '$.jobName') in {tuple(all_tasks_list)} ;
                """

    update_sql = f"""

                update
                    asw_etl_kvn_job.XX_SCHEDULE_JOB
                set
                    job_name = upper(replace(JSON_EXTRACT(PARAM_JSON, '$.jobName'),"\\"", ""))
                WHERE
                    PARAM_JSON like '%cdc-kvn-outbound-stream-crmbo%'
                    and JSON_EXTRACT(PARAM_JSON, '$.jobName') in {tuple(all_tasks_list)} ;
                """
    print(query_sql)
    # print(update_sql)


def run(bu):
    # 获取所有正在运行的任务
    url_stream_crmbo = f"https://cdc-api.ms.eu.aswatson.net/flink/cdc/cdc-{bu}-outbound-stream-crmbo/jobs/overview"
    all_tasks_dict = get_running_jobs(bu, url_stream_crmbo)
    # all_tasks_list = list(all_tasks_dict.keys())
    # print(all_tasks_list)

    # 处理监控的任务
    deal_monitor_task(bu, all_tasks_dict, url_stream_crmbo)

    # 处理重启的任务
    deal_restart_task(all_tasks_dict, bu)

    # 批量拼接SQl
    # genert_sql(bu,all_tasks_list)


if __name__ == '__main__':
    # bu = 'kvb'
    # run(bu)
    bu_list = ['kvn', 'kvb', 'wtctr', 'drlv', 'drlt', 'tps', 'icinl', 'icibe']
    for bu in bu_list:
        run(bu)
    pass
